摘要
這篇文章深入探討了如何將知識圖譜與 LangGraph 結合，打造一個智慧化的工安監控管理系統。文章首先介紹了知識圖譜在管理工安資料的優勢，包括儲存結構化和非結構化資訊、捕捉資料間的複雜關係，以及支援更深入的分析。然後，文章示範了如何使用 Neo4j 構建知識圖譜，並將工安事件資料導入其中。文章的核心部分在於利用 LangGraph 編織一個智能工作流，這個工作流包含了問題路由器、圖形查詢處理器和向量搜索處理器，能根據問題的類型選擇最合適的查詢方式，並提供更精確、更豐富的答案。文章最後展示了如何使用 LangGraph 來互動並執行這個智能工作流，並強調了知識圖譜、LangGraph 和檢索增強生成 (RAG) 的完美結合，為工地安全監控帶來了革命性的變革。

前言

在這個資訊爆炸的時代,我們該如何有效管理和運用海量的工安資料?今天,讓我們深入探討一個結合知識圖譜和語言模型的創新解決方案。

為什麼選擇知識圖譜?

傳統的 RAG (檢索增強生成) 系統主要處理非結構化文本,但在實際應用中,我們常常需要處理更複雜的資料關係。試想,如果我們能將工地位置、事故類型、相關人員和即時監控數據等多面向資訊串連起來,是否能讓我們的安全管理更上一層樓?

知識圖譜正是為此而生。它不僅能儲存結構化和非結構化資訊,還能捕捉資料之間的複雜關係。這意味著我們可以回答更深入的問題,例如「為何某類事故在特定地點頻繁發生?」或「如何評估預防措施的有效性?

工地安全監控知識圖譜：安全管理的RAG應用

節點與關係是描述知識圖譜中資料的基本元素。一般而言,節點用來表示實體或概念,如人員、地點和事件。在我們的工地安全監控圖譜中,節點描述了工人、監控人員、事故地點、事故類型及異常事件等。而關係則定義了這些實體之間的連結,例如事故發生的地點、異常事件是由哪一支攝影機記錄,或是由誰來處理等。

節點與關係都可以擁有以鍵值對形式儲存的屬性值。舉例來說,事故地點節點有名稱和地址兩個屬性,而事故節點則更為複雜,包含了ID、日期及嚴重程度等屬性。特別值得一提的是,我們可以在異常事件節點中加入文本嵌入值作為節點屬性。如此一來,我們就可以對異常事件的描述進行向量相似度搜尋,彷彿這些事件被儲存在向量資料庫中。因此,知識圖譜讓我們能夠同時儲存與檢索結構化和非結構化資訊,為RAG應用提供強大的支援。

這種結構不僅提供了豐富的上下文資訊,還支援複雜的推理。例如,我們可以輕鬆追蹤特定類型的事故在不同地點的發生頻率,或分析某位監控人員處理異常事件的效率。這大幅提升了RAG系統的分析深度,使其能夠回答更深入的問題,如「為何某類事故頻繁發生」或「如何有效預防特定類型的異常事件」。

關聯性分析

知識圖譜的核心優勢在於其強大的關聯分析能力。透過連接不同類型的節點,我們可以快速識別潛在的風險模式。例如:

MATCH (i:Incident)-[:CATEGORIZED_AS]->(t:IncidentType {name: '墜落'})
MATCH (i)-[:OCCURRED_AT]->(l:IncidentLocation)
MATCH (w:InvolvedPerson)-[:INVOLVED_IN]->(i)
RETURN i.id, i.description, w.name, l.name

這個查詢可以幫助幫助管理者了解具體事故的背景資訊及涉及人員，進而進行風險管理和決策。

構建知識圖譜:Neo4j 環境設置

此次資料及已經先放置於 Gist 當中，直接對接 Neo4j 進行操作就行。

節點：
- 事故（Incident）： 描述工地上發生的安全事故。
- 涉及人員（InvolvedPerson）： 工人或其他與事故有關的個體。
- 事故類型（IncidentType）： 事故的分類，如“跌倒”或“設備故障”。
- 事故地點（IncidentLocation）： 事故發生的具體地點。
- 監控攝像頭（SurveillanceCamera）： 用於監控工地各區域的攝像設備。
- 工地區域（SiteArea）： 監控的特定工地區域。
- 異常事件（AnomalyEvent）： 監控系統檢測到的異常活動，如未佩戴安全裝備的工人或進入危險區域。
- 監控人員（Monitor）： 負責監控系統操作和反應的工作人員。

邊：
- 事故發生在特定地點（OCCURRED_AT）： 將事故與具體的工地位置聯繫起來。
- 人員參與或受傷（INVOLVED_IN）： 表示與事故有關的工人或其他個體。
- 事故屬於某類型（CATEGORIZED_AS）： 為事故標記類型。
- 攝像頭監控特定區域（MONITORS）： 記錄監控攝像頭所覆蓋的工地區域。
- 異常事件被監控攝像頭記錄（RECORDED_BY）： 監控系統檢測到並記錄的異常事件。
- 異常事件與事故關聯（RELATED_TO）： 當異常事件與之後發生的事故有關時，將它們聯繫起來。
- 監控人員處理異常事件（HANDLED_BY）： 記錄監控人員如何處理監控到的異常情況。

應用情境：

1. 實時監控工地情況，結合過去的事故數據來分析高風險區域和危險行為。
2. 當異常事件被監控系統檢測到時，自動生成報告並與過去的事故數據進行比對，預測可能發生的安全問題，並及時處理。
3. 透過關聯異常事件和實際發生的事故，優化監控系統和安全管理措施，從而降低事故發生率。

聲明：資料完全由生成式 AI ChatGPT 建立，當中地名、人名、場域如有雷同，純屬巧合。

接下來,我們要將工安事件資料導入到我們的知識圖譜中。這個過程會將各種事件、地點和人員等資訊轉化為圖形中的節點和關係。

import requests
import_url = "https://gist.githubusercontent.com/Heng-xiu/a6417c4c34c80147cfbccdaa52925a55/raw/1126ea63fe28838d20af23441c4900ec768cfeb5/construction_site_safety.json"
import_query = requests.get(import_url).json()['query']
graph.query(import_query)

這段程式碼從遠端獲取預先準備好的 Cypher 查詢,並執行它來填充我們的圖形資料庫。

查詢圖形的兩種方式

在構建了知識圖譜之後,我們如何從中獲取有價值的信息呢?這裡有兩種主要的方法:

1. Cypher 查詢:直接使用 Neo4j 的查詢語言
2. 向量搜索:利用語義相似性進行檢索

向量檢索

讓我們先來看看向量檢索的實現:

import os
from langchain_community.vectorstores.neo4j_vector import Neo4jVector
from langchain_openai import OpenAIEmbeddings

safety_incidents_vector_index = Neo4jVector.from_existing_graph(
    OpenAIEmbeddings(),
    index_name='safety_incidents',
    node_label="Incident",
    text_node_properties=['id', 'date', 'severity', 'description'],
    embedding_node_property='embedding',
)

創建了一個向量索引,讓我們能夠基於語義相似性來搜索安全事件。現在,讓我們試試看:

response = safety_incidents_vector_index.similarity_search(
    "最近有沒有工人中暑的事故？"
)
print(response[0].page_content)

透過這種方式,我們可以快速找到與特定問題相關的事件,即使問題的表述與資料庫中的描述不完全一致。

Cypher 查詢

那麼,如果我們想要進行更複雜的查詢呢?這就是 Cypher 查詢發揮作用的時候了:

from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI

graph.refresh_schema()

cypher_chain = GraphCypherQAChain.from_llm(
    cypher_llm = ChatOpenAI(temperature=0, model_name='gpt-4o-mini'),
    qa_llm = ChatOpenAI(temperature=0),
    graph=graph,
    verbose=True,
    validate_cypher= True,
    return_direct = True,
)

cypher_chain.invoke(
    {"query": "有多少起高空墜落發生?"}
)

這個 GraphCypherQAChain 能夠將自然語言問題轉換為 Cypher 查詢,然後執行並解釋結果。這為我們提供了極大的靈活性,讓我們能夠提出各種複雜的問題。

結合 LangGraph:打造智能工作流

在我們的工安監控管理系統中,LangGraph 扮演了關鍵角色。它協助我們將各個組件串聯成一個智慧化的工作流程,讓系統能夠自動處理各種複雜的查詢。讓我們深入了解這個工作流的核心組成部分。

定義系統狀態：追蹤查詢進程

首先,我們需要定義一個狀態類來追蹤整個查詢過程。這個狀態類是我們工作流的基礎,它確保了資訊能夠在不同節點間順暢傳遞。

from typing import List, TypedDict

class SafetyIncident(TypedDict):
    id: str
    date: str
    severity: str
    description: str

class GraphState(TypedDict):
    question: str
    incidents: List[SafetyIncident]
    subqueries: object

這個 GraphState 類允許我們在整個查詢過程中保持狀態的一致性,確保資訊能夠順利地在不同節點間傳遞。

實作節點函數：系統的大腦

節點函數是我們工作流的核心處理單元。讓我們來看看幾個關鍵的節點函數：

問題路由器：智慧分流器

def route_question(state: GraphState):
    print("---ROUTE QUESTION---")
    question = state["question"]
    source = question_router.invoke({"question": question})
    if source.datasource == "vector search":
        print("---ROUTE QUESTION TO VECTOR SEARCH---")
        return "pass to vector search"
    elif source.datasource == "graph query":
        print("---ROUTE QUESTION TO GRAPH QA---")
        return "pass to graph query"

這個函數決定了每個問題應該走向哪條處理路徑。

圖形查詢處理器

def graph_qa(state: GraphState):
    question = state["question"]
    graph_qa_chain = GraphCypherQAChain.from_llm(
        cypher_llm = llm,
        qa_llm = llm,
        graph=graph,
        verbose=True,
        cypher_prompt=CYPHER_GENERATION_PROMPT,
        return_direct = True,
    )
    result = graph_qa_chain.invoke({"query": question})
    return {"documents": result, "question":question}

這個函數負責處理需要在知識圖譜中查詢的問題，它能夠將自然語言轉換為 Cypher 查詢，並解釋查詢結果。

向量搜索處理器：相似度探測器

def vector_search(state: GraphState):
    question = state["question"]
    queries = state["subqueries"]
    vector_graph_chain = RetrievalQA.from_chain_type(
        llm,
        chain_type="stuff",
        retriever = safety_incidents_vector_index.as_retriever(search_kwargs={'k':3}),
        verbose=True,
        return_source_documents=True,
    )
    chain_result = vector_graph_chain.invoke({"query": queries[0].sub_query})
    # ... 處理結果 ...
    return {
        "incident_ids": incident_ids,
        "incidents": extracted_data,
        "question": question,
        "subqueries": queries,
        "answer": chain_result['result']
    }

這個函數負責在向量空間中搜索相似的安全事故，它能夠處理那些需要語義理解的問題。

編織工作流：LangGraph 的精髓

有了這些組件，我們就可以使用 LangGraph 來編織我們的智能工作流了：

from langgraph.graph import END, StateGraph

workflow = StateGraph(GraphState)

# 添加節點
workflow.add_node(GRAPH_QA, graph_qa)
workflow.add_node(DECOMPOSER, decomposer)
workflow.add_node(VECTOR_SEARCH, vector_search)

# 設置條件入口點
workflow.set_conditional_entry_point(
    route_question,
    {
        'pass to vector search': DECOMPOSER,
        'pass to graph query': GRAPH_QA
    },
)

# 添加邊
workflow.add_edge(DECOMPOSER, VECTOR_SEARCH)
workflow.add_edge(VECTOR_SEARCH, END)
workflow.add_edge(GRAPH_QA, END)

app = workflow.compile()

這個工作流程的設計非常巧妙。它首先使用 route_question 函數來決定問題的處理方式。如果需要向量搜索,問題會被送到 DECOMPOSER 節點進行分解,然後再到 VECTOR_SEARCH 節點進行搜索。如果需要圖形查詢,問題則直接進入 GRAPH_QA 節點。
讓我們來想像一下這個工作流程在實際運作時的情景:

1. 當一個安全管理員問「最近有沒有類似於高空作業的安全事故?」時,系統會將其識別為需要向量搜索的問題。
2. 問題會被送到 DECOMPOSER 節點,可能會被分解為「找出最近的安全事故」和「判斷哪些事故與高空作業相似」。
3. 然後,VECTOR_SEARCH 節點會使用這些子查詢在向量空間中尋找相關的事故報告。

相反,如果問題是「哪個工地區域發生的高嚴重性事故最多?」,系統會將其識別為需要圖形查詢。這個問題會直接進入 GRAPH_QA 節點,在那裡它會被轉換為 Cypher 查詢並在知識圖譜中執行。

與工作流互動：系統的實際運作

現在，讓我們來看看如何與這個智慧工作流互動

vector_search_result = app.invoke({"question": "找出類似於密閉空間作業的安全協議"})
vector_search_result

這行程式碼會觸發整個工作流。系統會自動決定使用向量搜索來處理這個問題，然後返回相關的安全協議。
我們還可以查看結果的具體內容：

vector_search_result['incidents']

總結

在這篇探討現代工安管理的文章中，我們深入介紹了一個創新的系統，它巧妙地結合了檢索增強生成（RAG）、LangGraph 和知識圖譜技術，為工地安全監控帶來了革命性的變革。

首先，我們探討了知識圖譜如何能夠同時處理結構化和非結構化的工安數據，提供了一個全面的資訊網絡，使我們能夠回答更深入、更複雜的安全相關問題。接著，我們介紹了如何使用 Neo4j 建立和管理這個知識圖譜，以及如何將工安事件數據導入其中。

文章的核心部分詳細說明了如何利用 LangGraph 構建一個智能工作流。我們定義了系統狀態來追蹤查詢進程，實作了關鍵的節點函數如問題路由器、圖形查詢處理器和向量搜索處理器。這些組件共同構成了系統的「大腦」，能夠靈活處理各種工安相關查詢。

即刻前往教學程式碼 Repo，親自動手體驗知識圖譜結合 AI 代理的魅力吧！別忘了給專案按個星星並持續關注更新，讓我們一起探索AI代理的新境界。

x.參考資料

1. neo4j-graphrag